package com.ruoyi.eqmonitor.utils;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.ruoyi.common.core.exception.ServiceException;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.common.security.utils.DictUtils;
import com.ruoyi.eqmonitor.dispose.domain.OpentsdbTags;
import com.ruoyi.eqmonitor.dispose.dtos.OpentsdbTagsData;
import com.ruoyi.eqmonitor.domain.OptDeviceMonitorLastData;
import com.ruoyi.eqmonitor.dtos.OptDeviceDto;
import com.ruoyi.eqmonitor.service.IOptDeviceService;
import com.ruoyi.system.api.domain.SysDictData;
import lombok.Getter;
import lombok.extern.log4j.Log4j2;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.pool.SessionDataSetWrapper;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.opentsdb.client.bean.response.QueryResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

@Getter
@Log4j2
@Service
public class IotUtils {
    @Resource
    private SessionPool sessionPool;

    private static final Logger dataLogger = LoggerFactory.getLogger("monitor_data");
    
    @Lazy
    @Autowired
    private IOptDeviceService iOptDeviceService;

    @Value("${spring.iotdb.storeGroup}")
    private String storeGroup;

    public String getStoreGroup(String deviceCode) {
        return storeGroup.concat(".D").concat(deviceCode);
    }

    /**
     * 插入一条数据
     *
     * @param deviceId     设备id
     * @param time         时间
     * @param measurements 监测指标
     * @param types        监测指标值类型
     * @param values       监测值
     */
    public void insertRecord(String deviceId, long time, List<String> measurements, List<TSDataType> types, List<Object> values, String version) {
        try {
            dataLogger.info("IotDB数据入库：version: [{}], device_id:[{}], measurements:[{}], values:[{}]", version, deviceId, measurements, values);
            sessionPool.insertRecord(getStoreGroup(deviceId), time, measurements, types, values);
        } catch (Exception e) {
            dataLogger.error("IotDBSession insertRecord失败: version: {}, deviceId={}, time={}, measurements={}, values={}, error={}",
                    version, deviceId, time, measurements, values, e.getMessage());
        }
    }

    public void insertAlignedRecord(String deviceId, long time, List<String> measurements, List<TSDataType> types, List<Object> values, String version) {
        try {
            String sg = getStoreGroup(deviceId);
            dataLogger.info("IotDB数据入库：version: [{}], device_id:[{}], measurements:[{}], values:[{}]", version, deviceId, measurements, values);
            sessionPool.insertAlignedRecord(sg, time, measurements, types, values);
            //sessionPool.executeNonQueryStatement("flush");
        } catch (Exception e) {
            dataLogger.error("IotDBSession insertRecord失败: version: {}, deviceId={}, time={}, measurements={}, values={}, error={}",
                    version, deviceId, time, measurements, values, e.getMessage());
        }
    }

    public void write(OptDeviceMonitorLastData point, String version, String insertDeviceCode) {
        List<String> measurements = ListUtil.of(point.getCode());
        List<Object> values = ListUtil.of(point.getCodeValue());
        List<TSDataType> types = ListUtil.of(TSDataType.DOUBLE);
        this.insertAlignedRecord(
                insertDeviceCode,
                point.getMonitorTime().getTime(),
                measurements,
                types,
                values,
                version
        );
    }

    /**
     * 批量插入数据
     *
     * @param pointList        监测点列表
     * @param version          版本
     * @param insertDeviceCode 设备编码
     */
    public void write(List<OptDeviceMonitorLastData> pointList, String version, String insertDeviceCode) {
        long start = System.currentTimeMillis();
        dataLogger.info("IotDB数据入库：version: [{}], device_id:[{}], values: {}", version, insertDeviceCode, JSON.toJSONString(pointList));
        List<Long> times = new ArrayList<>();
        List<List<String>> measurementsList = new ArrayList<>();
        List<List<TSDataType>> typesList = new ArrayList<>();
        List<List<Object>> valuesList = new ArrayList<>();

        for (OptDeviceMonitorLastData point : pointList) {
            times.add(point.getMonitorTime().getTime());
            List<String> measurements = ListUtil.of(point.getCode());
            measurementsList.add(measurements);
            List<Object> values = ListUtil.of(point.getCodeValue());
            valuesList.add(values);
            List<TSDataType> types = ListUtil.of(TSDataType.DOUBLE);
            typesList.add(types);
        }

        try {
            String sg = getStoreGroup(insertDeviceCode);
            sessionPool.insertAlignedRecordsOfOneDevice(
                    sg,
                    times,
                    measurementsList,
                    typesList,
                    valuesList
            );
        } catch (IoTDBConnectionException | StatementExecutionException e) {
            dataLogger.error("IotDBSession insertRecord失败: {} \n version: {}, deviceId={}, values={}",
                    e.getMessage(), version, insertDeviceCode, JSON.toJSONString(pointList));
        }
        dataLogger.info("IotDB数据入库完成：version: [{}], device_id:[{}], 耗时: {}", version, insertDeviceCode, (System.currentTimeMillis() - start));
    }

    /**
     * 根据时间范围查询数据
     *
     * @param tableNames 库名,可以多个 例如：021YWD(设备编码)
     * @param columns    列名，可以多个 例如：0x0165(监测指标)
     * @param startTime  开始时间
     * @param endTime    结束时间
     * @return 查询结果集
     */
    public <T> Map<String, LinkedHashMap<Long, T>> queryList(List<String> tableNames, List<String> columns, Date startTime, Date endTime, Class<T> clazz) {
        return queryList(tableNames, columns, startTime.getTime(), endTime.getTime(), clazz);
    }

    /**
     * 根据时间范围查询数据
     *
     * @param tableNames 库名,可以多个 例如：021YWD(设备编码)
     * @param columns    列名，可以多个 例如：0x0165(监测指标)
     * @param startTime  开始时间
     * @param endTime    结束时间
     * @return 查询结果集
     */
    public <T> Map<String, LinkedHashMap<Long, T>> queryList(List<String> tableNames, List<String> columns, long startTime, long endTime, Class<T> clazz) {
        IotSqlBuilder builder = new IotSqlBuilder();
        if (CollectionUtil.isEmpty(columns)) {
            builder.selectAll();
        } else {
            builder.select(ArrayUtil.toArray(columns, String.class));
        }
        builder.from(ArrayUtil.toArray(tableNames, String.class))
                .where(new IotSqlBuilder.WhereBuilder().between("time", startTime, endTime));
        return query(builder, clazz);
    }

    /**
     * 根据时间范围查询数据
     *
     * @param tableNames 库名,可以多个 例如：021YWD(设备编码)
     * @param columns    列名，可以多个 例如：0x0165(监测指标)
     * @param startTime  开始时间
     * @param endTime    结束时间
     * @return 查询结果集
     */
    public <T> Map<String, LinkedHashMap<Long, T>> queryList(List<String> tableNames, List<String> columns, long startTime, long endTime, int pageNo, int pageSize, Class<T> clazz) {
        IotSqlBuilder builder = new IotSqlBuilder();
        if (CollectionUtil.isEmpty(columns)) {
            builder.selectAll();
        } else {
            builder.select(ArrayUtil.toArray(columns, String.class));
        }
        builder.from(ArrayUtil.toArray(tableNames, String.class))
                .where(new IotSqlBuilder.WhereBuilder().between("time", startTime, endTime))
                .page(pageNo, pageSize);
        return query(builder, clazz);
    }

    /**
     * 使用 IotSqlBuilder 构建查询
     * @param builder sql创建器
     * @param clazz 返回值类型
     * @return 查询结果集
     * @param <T> 返回值类型
     */
    public <T> Map<String, LinkedHashMap<Long, T>> query(IotSqlBuilder builder, Class<T> clazz) {
        return this.query(builder.build(getStoreGroup()), clazz);
    }

    /**
     * 使用sql语句构建查询
     *
     * @param sql   sql语句
     * @param clazz 返回值类型
     * @param <T>   返回值类型
     * @return 查询结果集
     */
    public <T> Map<String, LinkedHashMap<Long, T>> query(String sql, Class<T> clazz) {
        if (dataLogger.isDebugEnabled()) {
            dataLogger.debug("IotDB查询数据：sql: [{}]", sql);
        }
        try (SessionDataSetWrapper wrapper = sessionPool.executeQueryStatement(sql, 6000)) {
            wrapper.setBatchSize(1024);
            return handlerData(sql, wrapper, clazz);
        } catch (IoTDBConnectionException e) {
            throw new ServiceException("IotDB连接异常", e);
        } catch (StatementExecutionException e) {
            throw new ServiceException("IotDB执行异常", e);
        }
    }

    /**
     * 处理查询结果
     *
     * @param sql     sql语句
     * @param wrapper 查询结果
     * @param clazz   返回值类型
     * @param <T>     返回值类型
     * @return 查询结果集
     */
    private <T> Map<String, LinkedHashMap<Long, T>> handlerData(String sql, SessionDataSetWrapper wrapper, Class<T> clazz) throws IoTDBConnectionException, StatementExecutionException {
        Map<String, LinkedHashMap<Long, T>> resultMap = new LinkedHashMap<>();
        String groupName = getStoreGroup();
        groupName = groupName.concat(".");
        while (wrapper.hasNext()) {
            RowRecord next = wrapper.next();
            long timestamp = next.getTimestamp();
            List<Field> fields = next.getFields();
            List<String> columnNames = wrapper.getColumnNames();
            if (StrUtil.containsIgnoreCase(sql, "last")) {
                String column = fields.get(0).getStringValue(); // key
                column = handlerRemoveColumnStr(column, groupName);
                String value = fields.get(1).getStringValue(); // value
                if (StrUtil.contains(groupName, column)) {
                    column = StrUtil.removeAll(column, groupName);
                }
                LinkedHashMap<Long, T> map = resultMap.computeIfAbsent(column, k -> new LinkedHashMap<>());
                map.put(timestamp, "null".equals(value) ? null : Convert.convert(clazz, value));
            } else {
                // 不是聚合, columnNames[0]是Time, 会比fields多一个Time, 聚合的时候又没有Time字段了...
                int addSize = columnNames.size() - fields.size();
                for (int i = 0; i < columnNames.size() - 1; i++) {
                    String column = columnNames.get(i + addSize);
                    column = handlerRemoveColumnStr(column, groupName);
                    String value = fields.get(i).getStringValue();
                    LinkedHashMap<Long, T> map = resultMap.computeIfAbsent(column, k -> new LinkedHashMap<>());
                    map.put(timestamp, "null".equals(value) ? null : Convert.convert(clazz, value));
                }
            }
        }
        return resultMap;
    }

    /**
     * 处理列名
     *
     * @param column    列名
     * @param removeStr 需要移除的字符串
     * @return 处理后的列名
     */
    private String handlerRemoveColumnStr(String column, String removeStr) {
        if (StrUtil.containsIgnoreCase(column, removeStr)) {
            return StrUtil.removeAll(column, removeStr);
        }
        return column;
    }

    /**
     * 兼容opentsdb代码 构建QueryResult查询结果 推荐 不含额外查询]
     *
     * @param query 查询结果
     * @return 查询结果集
     */
    public List<QueryResult> buildOpentsdbQueryResult(Map<String, LinkedHashMap<Long, Number>> query) {
        return buildOpentsdbQueryResult(query, null);
    }

    /**
     * 兼容opentsdb代码 构建QueryResult查询结果 推荐 不含额外查询]
     *
     * @param query 查询结果
     * @param cb 回调自行注入部分信息的函数
     * @return 查询结果集
     */
    public List<QueryResult> buildOpentsdbQueryResult(Map<String, LinkedHashMap<Long, Number>> query, Callback<QueryResult> cb) {
        return query.entrySet().stream().map(entry -> {
            String key = entry.getKey();
            // 判断key被()包裹的情况
            if (StrUtil.contains(key, "(") && StrUtil.contains(key, ")")) {
                key = StrUtil.subBetween(key, "(", ")");
            }
            String snCode = StrUtil.subBefore(key, ".", false);
            String monitorCode = StrUtil.subAfter(key, ".", false);
            QueryResult queryResult = new QueryResult();
            queryResult.setMetric(key);
            HashMap<String, String> tags = new HashMap<>();
            tags.put("snCode", snCode);
            tags.put("monitorCode", monitorCode);
            queryResult.setTags(tags);
            queryResult.setAggregateTags(new ArrayList<>());
            queryResult.setDps(entry.getValue());
            if (cb != null) {
                cb.call(snCode, monitorCode, queryResult);
            }
            return queryResult;
        }).collect(Collectors.toList());
    }

    /**
     * 兼容opentsdb代码 构建QueryResult查询结果 [会返回设备类型、监测指标名称、单位等信息 因为需要多次查询 所以效率不高 不需要详情不推荐使用]
     *
     * @param query 查询结果
     * @return 查询结果集
     */
    public List<QueryResult> buildOpentsdbQueryResultAndDetail(Map<String, LinkedHashMap<Long, Number>> query) {
        return buildOpentsdbQueryResultAndDetail(query, null);
    }

    /**
     * 兼容opentsdb代码 构建QueryResult查询结果 [会返回设备类型、监测指标名称、单位等信息 因为需要多次查询 所以效率不高 不需要详情不推荐使用]
     *
     * @param query 查询结果
     * @param cb 回调自行注入部分信息的函数
     * @return 查询结果集
     */
    public List<QueryResult> buildOpentsdbQueryResultAndDetail(Map<String, LinkedHashMap<Long, Number>> query, Callback<QueryResult> cb) {
        CacheInfo cacheInfo = new CacheInfo(iOptDeviceService);
        return this.buildOpentsdbQueryResult(query, (snCode, monitorCode, result) -> {
            Map<String, String> tags = result.getTags();
            if (tags == null) {
                tags = new HashMap<>();
                result.setTags(tags);
            }
            // 监测指标相关
            if (StrUtil.isNotBlank(monitorCode)) {
                Constant.MonitorInfo monitorInfo = cacheInfo.getMonitorInfoCache(monitorCode);
                if (monitorInfo != null) {
                    tags.put("monitorName", monitorInfo.getName());
                    tags.put("dataUnit", monitorInfo.getUnit());
                }
            }
            // 设备相关
            if (StrUtil.isNotBlank(snCode)) {
                OptDeviceDto device = cacheInfo.getDeviceCache(snCode);
                if (device != null) {
                    String deviceType = device.getDeviceType();
                    tags.put("equipmentType", deviceType);
                    tags.put("equipmentTypeCode", cacheInfo.getDeviceTypeCache(deviceType));
                }
            }
            if (cb != null) {
                cb.call(snCode, monitorCode, result);
            }
        });
    }

    /**
     * 反序列化opentsdb时序日期单位
     * @param timestamp opentsdb时序时间
     * @return 反序列化日期单位
     */
    public static String deserializeDate(Long timestamp) {
        return DateUtil.format(new Date(timestamp), "yyyy-MM-dd");
    }

    /**
     * 反序列化opentsdb时序时间单位 不含秒
     * @param timestamp opentsdb时序时间
     * @return 反序列化时间单位
     */
    public static String deserializeTime(Long timestamp) {
        return DateUtil.format(new Date(timestamp), "HH:mm");
    }

    /**
     * 反序列化opentsdb时序日期时间单位 'yyyy-MM-dd HH:mm:ss'
     * @param timestamp opentsdb时序时间
     * @return 反序列化时间单位
     */
    public static String deserialize(Long timestamp) {
        return DateUtil.formatDateTime(new Date(timestamp));
    }

    /**
     * 兼容opentsdb代码 构建OpentsdbTagsData查询结果 推荐使用 不含额外查询]
     * @param query 查询结果
     * @return 查询结果集
     */
    public List<OpentsdbTagsData> buildOpentsdbTags(Map<String, LinkedHashMap<Long, Number>> query) {
        return buildOpentsdbTags(query, null);
    }

    /**
     * 兼容opentsdb代码 构建OpentsdbTagsData查询结果 推荐使用 不含额外查询]
     * @param query 查询结果
     * @param cb 回调自行注入部分信息的函数
     * @return 查询结果集
     */
    public List<OpentsdbTagsData> buildOpentsdbTags(Map<String, LinkedHashMap<Long, Number>> query, Callback<OpentsdbTagsData> cb) {
        return query.entrySet().stream().map(entry -> {
            String key = entry.getKey();
            // 判断key被()包裹的情况
            if (StrUtil.contains(key, "(") && StrUtil.contains(key, ")")) {
                key = StrUtil.subBetween(key, "(", ")");
            }
            LinkedHashMap<Long, Number> map = entry.getValue();
            String snCode = StrUtil.subBefore(key, ".", false);
            String monitorCode = StrUtil.subAfter(key, ".", false);
            OpentsdbTags opentsdbTags = new OpentsdbTags();
            opentsdbTags.setSnCode(snCode);
            opentsdbTags.setMonitorCode(monitorCode);
            OpentsdbTagsData result = new OpentsdbTagsData(opentsdbTags, map);
            if (cb != null) {
                cb.call(snCode, monitorCode, result);
            }
            return result;
        }).collect(Collectors.toList());
    }

    /**
     * 兼容opentsdb代码 构建OpentsdbTagsData查询结果 [会返回设备类型、监测指标名称、单位等信息 因为需要多次查询 所以效率不高 不需要详情不推荐使用]
     * @param query 查询结果
     * @return 查询结果集
     */
    public List<OpentsdbTagsData> buildOpentsdbTagsAndDetail(Map<String, LinkedHashMap<Long, Number>> query) {
        return buildOpentsdbTagsAndDetail(query, null);
    }

    /**
     * 兼容opentsdb代码 构建OpentsdbTagsData查询结果 [会返回设备类型、监测指标名称、单位等信息 因为需要多次查询 所以效率不高 不需要详情不推荐使用]
     * @param query 查询结果
     * @param cb 回调自行注入部分信息的函数
     * @return 查询结果集
     */
    public List<OpentsdbTagsData> buildOpentsdbTagsAndDetail(Map<String, LinkedHashMap<Long, Number>> query, Callback<OpentsdbTags> cb) {
        CacheInfo cacheInfo = new CacheInfo(iOptDeviceService);
        return buildOpentsdbTags(query, (snCode, monitorCode, result) -> {
            OpentsdbTags opentsdbTags = result.getOpentsdbTags();
            if (opentsdbTags == null) {
                opentsdbTags = new OpentsdbTags();
                result.setOpentsdbTags(opentsdbTags);
            }
            // 设备相关
            if (StrUtil.isNotBlank(snCode)) {
                OptDeviceDto device = cacheInfo.getDeviceCache(snCode);
                if (device != null) {
                    String deviceType = device.getDeviceType();
                    opentsdbTags.setEquipmentTypeCode(deviceType);
                    opentsdbTags.setEquipmentType(cacheInfo.getDeviceTypeCache(deviceType));
                }
            }
            // 监测指标相关
            if (StrUtil.isNotBlank(monitorCode)) {
                Constant.MonitorInfo monitorInfo = cacheInfo.getMonitorInfoCache(monitorCode);
                if (monitorInfo != null) {
                    opentsdbTags.setMonitorName(monitorInfo.getName());
                    opentsdbTags.setMonitorCode(monitorInfo.getCode());
                    opentsdbTags.setDataUnit(monitorInfo.getUnit());
                }
            }
            if (cb != null) {
                cb.call(snCode, monitorCode, opentsdbTags);
            }
        });
    }

    /**
     * 一个处理从iotDB数据转换为opentsdb数据的缓存类
     */
    private static class CacheInfo {
        private final Map<String, OptDeviceDto> deviceMap;
        private final Map<String, Constant.MonitorInfo> monitorInfoMap;
        private final Map<String, String> deviceTypeMap;
        private final IOptDeviceService iOptDeviceService;

        public CacheInfo(IOptDeviceService iOptDeviceService) {
            this.deviceMap = new HashMap<>();
            this.monitorInfoMap = new HashMap<>();
            this.iOptDeviceService = iOptDeviceService;
            List<SysDictData> dictCache = DictUtils.getDictCache("opt_device_type");
            if (dictCache == null) {
                dictCache = new ArrayList<>();
            }
            this.deviceTypeMap = dictCache.stream().collect(Collectors.toMap(SysDictData::getDictValue, SysDictData::getDictLabel));
        }

        public OptDeviceDto getDeviceCache(String snCode) {
            return this.deviceMap.computeIfAbsent(snCode, this.iOptDeviceService::getDeviceByCode);
        }

        public Constant.MonitorInfo getMonitorInfoCache(String monitorCode) {
            return this.monitorInfoMap.computeIfAbsent(monitorCode, Constant.MonitorInfo::getInfoByCode);
        }

        public String getDeviceTypeCache(String deviceType) {
            return this.deviceTypeMap.get(deviceType);
        }
    }

    /**
     * 回调函数
     * @param <T> 回调函数返回值类型
     */
    public interface Callback<T> {
        /**
         * 一个兼容opentsdb转换结果集的回调函数
         * @param snCode 设备编码
         * @param monitorCode 监测指标编码
         * @param data 回调函数处理的对象
         */
        void call(String snCode, String monitorCode, T data);
    }

    /**
     * 使用 IotSqlBuilder 构建查询(二次封装）
     * @param select 列名
     * @param fill 填充值 可使用 FillType 枚举 或 BOOLEAN, INT32, INT64, FLOAT, DOUBLE, TEXT
     * @param deviceCodes 表名
     * @param startTime 开始时间
     * @param endTime 结束时间
     * @param interval 时间间隔 (ms, s, m, h, d, mo, y) 例如：1d
     * @return 查询结果集
     */
    public List<OpentsdbTagsData> secondaryQuery(String[] select,
                                                 String fill,
                                                 String[] deviceCodes,
                                                 Date startTime,
                                                 Date endTime,
                                                 String interval) {
        if (startTime == null){
            //起始时间为当天0点时间，结束时间为当前时间
            String sdf = DateUtils.YYYYMMDD;
            endTime = new Date();
            startTime = DateUtils.dateTime(sdf,DateUtils.toString(endTime,sdf));
        }

        IotSqlBuilder iotSqlBuilder = new IotSqlBuilder()
                .select(select);
        if (fill != null){
            iotSqlBuilder.fill(fill);
        }
        iotSqlBuilder.from(deviceCodes).groupByTime(startTime, endTime, interval);
        Map<String, LinkedHashMap<Long, Number>> query = this.query(iotSqlBuilder, Number.class);
        return this.buildOpentsdbTags(query);
    }

}
